#ifndef __M_CHANNAL_H__
#define __M_CHANNAL_H__
#include "../mqCommon/logger.hpp"
#include "../mqCommon/helper.hpp"
#include "../mqCommon/mq_thread_pool.hpp"
#include "../mqCommon/mq_msg.pb.h"
#include "../mqCommon/mq_proto.pb.h"
#include "../thirdLib/net/TcpConnection.h"
#include "../thirdLib/proto/codec.h"
#include "../thirdLib/proto/dispatcher.h"
#include "mq_consumer.hpp"
#include "mq_virtual_host.hpp"
#include "mq_route.hpp"
#include "mq_message.hpp"
#include <iostream>


namespace yqx{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using openChannelRequestPtr = std::shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr = std::shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr = std::shared_ptr<deleteQueueRequest>;
    using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;
    using queueUnbindRequestPtr = std::shared_ptr<queueUnbindRequest>;
    using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;
    using basicConsumeRequestPtr = std::shared_ptr<basicConsumeRequest>;
    using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;

    class Channal{
        public:
            using ptr = std::shared_ptr<Channal>;
            Channal(const std::string& cid, const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec, const ConsumerManager::ptr& cmp, const VirtualHost::ptr& vhp, const ThreadPool::ptr pool):
            _cid(cid), _conn(conn), _codec(codec), _cmp(cmp), _vhp(vhp), _thread_pool(pool)
            {}
            ~Channal(){
                if(_csp != nullptr)
                    _cmp->remove(_csp->_qname, _csp->_tag);
            }
            //声明交换机
            void declareExchange(const declareExchangeRequestPtr& req){
                bool ret = _vhp->declareExchange(req->exchange_name(), req->exchange_type(), req->durable(), req->auto_delete());
                basicResponse(ret, req->rid(), req->cid());
            }
            //删除交换机
            void deleteExchange(const deleteExchangeRequestPtr& req){
                _vhp->deleteExchange(req->exchange_name());
                basicResponse(true, req->rid(), req->cid());
            }
            //声明队列
            void declareQueue(const declareQueueRequestPtr& req){
                bool ret = _vhp->declareQueue(req->queue_name(), req->durable(), req->exclusive(), req->auto_delete());
                if(ret == false)
                    basicResponse(false, req->rid(), req->cid());
                _cmp->initQueueConsumer(req->queue_name());
                basicResponse(true, req->rid(), req->cid());
            }
            //删除队列
            void deleteQueue(const deleteQueueRequestPtr& req){
                _vhp->deleteQueue(req->queue_name());
                _cmp->destroyQueueConsumer(req->queue_name());
                basicResponse(true, req->rid(), req->cid());
            }
            //绑定交换机和队列
            void bind(const queueBindRequestPtr& req){
                bool ret = _vhp->bind(req->exchange_name(), req->queue_name(), req->binding_key());
                basicResponse(ret, req->rid(), req->cid());
            }
            //解绑
            void unbind(const queueUnbindRequestPtr& req){
                _vhp->unbind(req->exchange_name(), req->queue_name());
                basicResponse(true, req->rid(), req->cid());
            }
            //发布消息
            void basicPublish(const basicPublishRequestPtr& req){
                //获取交换机信息
                auto ep = _vhp->getExchange(req->exchange_name());
                if(ep == nullptr) 
                    basicResponse(false, req->rid(), req->cid());
                //进行交换路由，判断消息被发送到交换机的哪个队列中去
                MsgQueueBindingMap mqbm = _vhp->getExchangeBindings(req->exchange_name());
                BasicProperties* bp = nullptr;
                std::string routing_key;
                if(req->has_properties()){
                    bp = req->mutable_properties();
                    routing_key = bp->routing_key();
                }
                for(auto& binding : mqbm){
                    //将消息添加到队列中
                    if(Router::route(ep->_type, routing_key, binding.second->_binding_key)){
                        _vhp->basicPublish(binding.first, bp, req->body());
                        //向线程池中添加一个消息消费的任务（向指定队列的订阅者推送消息）
                        auto task = std::bind(&Channal::consume, this, binding.first);
                        _thread_pool->push(task);
                    }
                        
                }
                basicResponse(true, req->rid(), req->cid());
            }
            //确认消息
            void basicAck(const basicAckRequestPtr& req){
                _vhp->basicAck(req->queue_name(), req->message_id());
                basicResponse(true, req->rid(), req->cid());
            }
            //订阅消息
            void basicConsume(const basicConsumeRequestPtr& req){
                //判断队列是否存在
                MsgQueue::ptr qp = _vhp->getQueue(req->queue_name());
                if(qp == nullptr){
                    LOG("订阅的队列<%s>不存在！", req->queue_name().c_str());
                    return basicResponse(false, req->rid(), req->cid());
                }
                //创建队列的消费者
                auto cb = std::bind(&Channal::callback, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
                //创建了消费者后，当前信道就是消费者了！
                _csp = _cmp->create(req->consumer_tag(), req->queue_name(), req->auto_ack(), cb);
                basicResponse(true, req->rid(), req->cid());
            }
            //取消订阅
            void basicCancle(const basicCancelRequestPtr& req){
                _cmp->remove(req->queue_name(), req->consumer_tag());
                basicResponse(true, req->rid(), req->cid());
            }
        private:
            //消费者的回调函数，一旦有消息发布到这个队列就会调用这个函数给消费者发送信息
            void callback(const std::string& tag, const BasicProperties* bp, const std::string& body){
                basicConsumeResponse resp;
                resp.set_cid(_cid);
                resp.set_body(body);
                resp.set_consumer_tag(tag);
                if(bp){
                    resp.mutable_properties()->set_id(bp->id());
                    resp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                    resp.mutable_properties()->set_routing_key(bp->routing_key());
                }
                _codec->send(_conn, resp);

            }
            void consume(const std::string qname){
                //从队列中取出一条消息
                MessagePtr mp = _vhp->basicConsume(qname);
                if(mp == nullptr){
                    LOG("执行消费任务失败！%s队列没有消息!", qname.c_str());
                    return;
                }
                //从队列订阅者中取出一个订阅者
                Consumer::ptr cp = _cmp->choose(qname);
                if(cp == nullptr){
                    LOG("执行消费任务失败！%s队列没有消费者!", qname.c_str());
                    return;
                }
                //调用订阅者对应的消息处理回调函数，实现消息的推送
                cp->_callback(cp->_tag, mp->mutable_payload()->mutable_properties(), mp->payload().body());
                //判断订阅者是否是自动确认——不需要等待确认就直接删除信息，否则需要外部受到信息确认后再删除
                if(cp->_auto_ack)
                    _vhp->basicAck(qname, mp->payload().properties().id());
            }
            void basicResponse(bool ok, const std::string& rid, const std::string& cid){
                basicCommonResponse resp;
                resp.set_ok(ok);
                resp.set_rid(rid);
                resp.set_cid(cid);
                _codec->send(_conn, resp);
            }
        
        private:
            std::string _cid;
            Consumer::ptr _csp;
            muduo::net::TcpConnectionPtr _conn;
            ProtobufCodecPtr _codec;
            ConsumerManager::ptr _cmp;
            VirtualHost::ptr _vhp;
            ThreadPool::ptr _thread_pool;
    };
    class ChannalManager{
        public:
            using ptr = std::shared_ptr<ChannalManager>;
            ChannalManager(){}
            //打开一个信道
            bool openChannal(const std::string& cid, const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec, const ConsumerManager::ptr& cmp, const VirtualHost::ptr& vhp, const ThreadPool::ptr pool){
                std::unique_lock<std::mutex> lock(_mtx);
                if(_channals.find(cid) != _channals.end()){
                    LOG("<%s>信道已经存在！", cid.c_str());
                    return false;
                }
                auto channal = std::make_shared<Channal>(cid, conn, codec, cmp, vhp, pool);
                _channals.insert(std::make_pair(cid, channal));
                return true;
            }
            //关闭一个信道
            void closeChannal(const std::string& cid){
                std::unique_lock<std::mutex> lock(_mtx);
                _channals.erase(cid);
            }
            //获取一个信道
            Channal::ptr getChannal(const std::string& cid){
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _channals.find(cid);
                if(it == _channals.end()){
                    LOG("<%s>信道不存在！", cid.c_str());
                    return nullptr;
                }
                return it->second;
            }

        private:
            std::mutex _mtx;
            std::unordered_map<std::string, Channal::ptr> _channals;

    };
}
#endif